#include "flow_inspector.h"
#include "base64.h"
#include "utils.h"
#include <functional>
#include <vector>
#include <utility>

using std::vector;

static const size_t REQ_INTERVAL = 500;
static const size_t INTERVAL = 60;
const std::string InspectorName("FlowInspector");

static const std::vector<std::pair<std::string, std::string>> dest_email = {
  {"yurunyang1998", "yurunyang1998@qq.com"},
  {"centrifugalwind", "centrifugalwind@163.com"}
};
static const std::string email_subject{"Inspector Alert! Service May Be DUMP!"};
static const char *smtp_server = "smtp.163.com";
static const char *username = "unicxitoiv@163.com";
static const char *passwd{"xxxxxx"};

FlowInspector::FlowInspector(EventLoop *loop, const InetAddress &servAddr)
  : loop_(loop)
  , client_(loop, servAddr, "InspectorName")
  , email_client_()
  , retry_count_(0) {
  loop->runEvery(REQ_INTERVAL, std::bind(&FlowInspector::reconnect, this));
  client_.setConnectionCallback(std::bind(&FlowInspector::OnConnection, this, _1));
  client_.setMessageCallback(std::bind(&FlowInspector::OnMessage, this, _1, _2, _3));
  client_.setRetryConnectCallback(std::bind(&FlowInspector::OnRetryConnect, this));
}

FlowInspector::~FlowInspector() { }

void FlowInspector::connect() {
  client_.connect();
}

void FlowInspector::reconnect() {
  client_.reconnect();
}

void FlowInspector::OnConnection(const TcpConnectionPtr& conn) {
  LOG_INFO << conn->localAddress().toIpPort() << " -> "
           << conn->peerAddress().toIpPort() << " is "
           << (conn->connected() ? "UP" : "DOWN");
  if (conn->connected()) {
    retry_count_ = 0;
    LOG_INFO << "connect balancer successfully!";
    string format_content =
      utils::format_string(http_content, base64_encode(utils::testing_account));
    conn->send(format_content.c_str(), format_content.length());
    loop_->runAfter(INTERVAL, std::bind(&FlowInspector::checkReceive, this, conn));
    activeConn_.insert({conn->name(), false});
  }
}

void FlowInspector::OnMessage(const TcpConnectionPtr& conn, Buffer *buf, Timestamp recvTime) {
  std::string buff(buf->peek());
  if (!IsValid(buff)) {
    LOG_WARN << "balancer: Service Unavailable!";
    Alert(atType_Unreachable);
  } else {
    LOG_INFO << "balancer is fine.";
  }
  activeConn_[conn->name()] = true;
  conn->shutdown();
}

void FlowInspector::OnRetryConnect() {
  LOG_WARN << "balancer may be dump!";
  retry_count_++;
  Alert(atType_Offline);
}

void FlowInspector::Alert(atType type) {
  email_client_ << smtp::server(smtp_server);
  email_client_ << smtp::user(username, passwd);
  email_client_ << smtp::from("inspector bot", username);
  for (const auto& iter : dest_email) {
    email_client_ << smtp::to(iter.first.c_str(), iter.second.c_str());
  }
  email_client_ << smtp::subject(email_subject.c_str());
  switch (type) {
  case atType_Offline:
    email_client_ << smtp::mimetext(
        utils::format_string("Can NOT Connect to balancer! Retry count: {}", retry_count_).c_str());
    break;
  case atType_Unreachable:
    email_client_ << smtp::mimetext("Balancer is connected, but the service is NOT available!");
    break;
  }
  email_client_ << smtp::directive::syncperform;
}

bool FlowInspector::IsValid(const string& buff) {
  string::size_type left_cur = buff.find("HTTP/1.1");
  if (left_cur == string::npos) {
    return false;
  }

  string::size_type right_cur = buff.find("\r\n", left_cur);
  if (right_cur == string::npos) {
    LOG_WARN << "receive uncomplete message";
    // ignore the error like this.
    return true;
  }

  auto result_split = utils::split_string(buff.substr(left_cur, right_cur - left_cur), " ");
  if (result_split.size() > 2) {
    if (result_split[1] == "401" || result_split[1] == "407") {
      LOG_WARN << "balancer work fine. But request account unavaliable!";
    }
    return true;
  }

  return false;
}

void FlowInspector::OnEmailSendComplete(const smtp::Request& req) {
  if (!req.error.empty()) {
    LOG_INFO << "Error: " << req.error;
  } else {
    LOG_INFO << "Send Email Success!";
  }
}

void FlowInspector::checkReceive(const TcpConnectionPtr& conn) {
  auto iter = activeConn_.find(conn->name());
  if (iter == activeConn_.end()) {
    LOG_ERROR << "The activity connection has been deleted from anywhere, "
              << "and it could not going here...";
    return;
  }

  if (!iter->second) {
    LOG_WARN << "balancer NOT reply!";
    Alert(atType_Unreachable);
  }

  activeConn_.erase(iter);
}

